package com.zx.learn.flink.sink;

import com.zx.learn.flink.utils.PathUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Description 将DataSet数据写入到csv文件中
 */
@Slf4j
public class CsvSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //需先建立文件
        String fullFilePathName = PathUtil.CLASSPATH_DATA_OUTPUT + "csv.sink.csv";
        log.debug("file : {}", fullFilePathName);
        //添加数据,writeAsCsv 只能Sink Tuple3
        //Tuple7<Integer, String, Integer, Integer, String, String, Long> row =
        // new Tuple7<>(15, "zhangsan", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L);
        Tuple3<Integer, String, Long> row = new Tuple3<>(15, "zhangsan", 1599494400000L);
        //转换为dataSet
        DataStream<Tuple3<Integer, String, Long>> dataSet = env.fromElements(row);
        //将内容写入到File中，如果文件已存在，将会被复盖
        dataSet.writeAsCsv(fullFilePathName, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        env.execute("fline file sink");
    }
}
